package com.cw.starter.mq.producer;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cw.starter.mq.dto.MqMsgDto;
import com.cw.starter.mq.listener.SendCallbackListener;

import cn.hutool.core.util.StrUtil;

public class MqMessageProducer {
    private static final Logger log = LoggerFactory.getLogger(MqMessageProducer.class);

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * rocketmq 同步消息
     *
     * @param topic 主题
     * @param mqMsgDto 消息
     * @return 结果
     */
    public SendResult send(String topic, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushMessage start : " + sendMessage);
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .build();
        // 设置发送地和消息信息并发送同步消息
        SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
        log.info("pushMessage finish : " + sendMessage + ", sendResult : " + JSONObject.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * rocketmq 同步消息
     *
     * @param topic 主题
     * @param syncTag 同步tag
     * @param mqMsgDto 消息
     * @return 结果
     */
    public SendResult sendMessage(String topic, String syncTag, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushMessage start : " + sendMessage);
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .build();
        // 设置发送地和消息信息并发送同步消息
        SendResult sendResult = rocketMQTemplate.syncSend(StrUtil.isBlank(syncTag)? topic : topic + ":" + syncTag, message);
        log.info("pushMessage finish : " + sendMessage + ", sendResult : " + JSONObject.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * 发送异步消息
     *
     * @param mqMsgDto 消息
     * @param topic 消息
     * @return 结果
     */
    public void sendAsyncMessage(String topic, String asyncTag, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushAsyncMessage start : " + sendMessage);
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .build();
        // 设置发送地和消息信息并发送异步消息
        rocketMQTemplate.asyncSend(StrUtil.isBlank(asyncTag)? topic : topic + ":" + asyncTag, message, new SendCallbackListener(mqMsgDto));
        log.info("pushAsyncMessage finish : " + sendMessage);
    }

    /**
     * 发送单向消息（不关注发送结果：记录日志）
     *
     * @param mqMsgDto 消息
     * @return 结果
     */
    public void sendOneWayMessage(String topic, String onewayTag, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushOneWayMessage start : " + sendMessage);
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .build();
        // 设置发送地和消息信息并发送单向消息
        rocketMQTemplate.sendOneWay(StrUtil.isBlank(onewayTag)? topic : topic + ":" + onewayTag, message);
        log.info("pushOneWayMessage finish : " + sendMessage);
    }

    /**
     * rocketmq 延迟消息
     *
     * @param mqMsgDto 消息
     * @return 结果
     */
    public SendResult sendDelayMessage(String topic, String syncTag, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushDelayMessage start : " + sendMessage);
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .build();
        // 设置超时和延时推送
        // 超时时针对请求broker然后结果返回给product的耗时
        // 现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
        // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        SendResult sendResult = rocketMQTemplate.syncSend(StrUtil.isBlank(syncTag)? topic : topic + ":" + syncTag, message, 1 * 1000L, 4);
        log.info("pushDelayMessage finish : " + sendMessage + ", sendResult : " + JSONObject.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * 事务消息
     *
     * @param mqMsgDto 消息
     * @return 结果
     */
    public TransactionSendResult sendTransactionMessage(String topic, String syncTag, MqMsgDto mqMsgDto) {
        String sendMessage = JSON.toJSONString(mqMsgDto);
        log.info("pushTransactionMessage start : " + JSON.toJSONString(sendMessage));
        // 创建消息
        Message<String> message = MessageBuilder.withPayload(sendMessage)
            .setHeader(RocketMQHeaders.TRANSACTION_ID,mqMsgDto.getSerialNo())
            .build();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(StrUtil.isBlank(syncTag)? topic : topic + ":" + syncTag,
            message, null);
        log.info("pushTransactionMessage result : " + JSONObject.toJSONString(transactionSendResult));
        log.info("pushTransactionMessage finish : " + JSON.toJSONString(sendMessage));
        return transactionSendResult;
    }
}